package com.xrui.hbase.impl;

import com.google.common.base.Objects;
import com.google.common.base.Preconditions;

import com.xrui.hbase.*;
import com.xrui.hbase.schema.TableLayout;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Map;
import java.util.NavigableMap;
import java.util.concurrent.atomic.AtomicReference;

public final class HBaseTableWriter implements TableWriter {
    //region static
    private static final Logger LOG = LoggerFactory.getLogger(HBaseTableWriter.class);
    /**
     * Tracks the state of this writer.
     */
    private final AtomicReference<State> mState = new AtomicReference<State>(State.UNINITIALIZED);
    //endregion

    //region variables
    /**
     * The hbase table instance.
     */
    private final HBaseTable mTable;
    /**
     * Layout consumer registration resource.
     */
    private final TableLayout mTableLayout;
    /**
     * Dedicated HTable connection.
     */
    private final HTableInterface mHTable;

    /**
     * Creates a non-buffered table writer that sends modifications directly to hbase.
     *
     * @param table A hbase table.
     * @throws IOException on I/O error.
     */
    public HBaseTableWriter(HBaseTable table) throws IOException {
        mTable = table;
        mTableLayout = table.getLayout();
        mHTable = table.openHTableConnection();

        Preconditions.checkState(mTableLayout != null,
            "TableWriter for table: %s failed to initialize.", mTable.getURI());

        //mHTable.setAutoFlush(true);

        // Retain the table only when everything succeeds.
        mTable.retain();
        final State oldState = mState.getAndSet(State.OPEN);
        Preconditions.checkState(oldState == State.UNINITIALIZED,
            "Cannot open TableWriter instance in state %s.", oldState);
    }
    //endregion

    //region construct

    /**
     * {@inheritDoc}
     */
    @Override
    public <T> void put(EntityId entityId, String family, String qualifier, T value)
        throws IOException {
        put(entityId, family, qualifier, HConstants.LATEST_TIMESTAMP, value);
    }
    //endregion

    //region override
    //region put

    /**
     * {@inheritDoc}
     */
    @Override
    public <T> void put(EntityId entityId, String family, String qualifier, long timestamp, T value)
        throws IOException {
        final State state = mState.get();
        Preconditions.checkState(state == State.OPEN,
            "Cannot put cell to TableWriter instance %s in state %s.", this, state);

        //final ColumnName columnName = ColumnName.create(family, qualifier);

        //TODO: wait for encode implement
        //final CellEncoder cellEncoder = CellEncoderProvider.getEncoder(
        //    mTableLayout, mTableLayout.getSchemaResolver(), GenericCellDecoderFactory.get(), Maps.newHashMap()
        //);
        //    capsule.getCellEncoderProvider().getEncoder(family, qualifier);
        //final byte[] encoded = cellEncoder.encode(value);

        final Put put = new Put(entityId.getHBaseRowKey())
            .add(family.getBytes(), qualifier.getBytes(), timestamp, "To Be implements".getBytes());
        mHTable.put(put);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public HBaseCell<Long> increment(EntityId entityId, String family, String qualifier, long amount)
        throws IOException {
        final State state = mState.get();
        Preconditions.checkState(state == State.OPEN,
            "Cannot increment cell to TableWriter instance %s in state %s.", this, state);

        //verifyIsCounter(family, qualifier);

        final ColumnName columnName = ColumnName.create(family, qualifier);
        // Send the increment to the HBase HTable.
        final Increment increment = new Increment(entityId.getHBaseRowKey());
        increment.addColumn(
            columnName.getFamilyBytes(),
            columnName.getQualifierBytes(),
            amount);
        final Result result = mHTable.increment(increment);
        final NavigableMap<Long, byte[]> counterEntries =
            result.getMap().get(columnName.getFamilyBytes()).get(columnName.getQualifierBytes());
        assert null != counterEntries;
        assert 1 == counterEntries.size();

        final Map.Entry<Long, byte[]> counterEntry = counterEntries.firstEntry();
        final DecodedCell<Long> counter = new DecodedCell<Long>(
            DecodedCell.NO_SCHEMA,
            Bytes.toLong(counterEntry.getValue()));
        return HBaseCell.create(entityId, columnName, counterEntry.getKey(), counter);
    }

    //endregion

    //region Incrementer

    /**
     * {@inheritDoc}
     */
    @Override
    public void deleteRow(EntityId entityId) throws IOException {
        deleteRow(entityId, HConstants.LATEST_TIMESTAMP);
    }
    //endregion

    //region Deleter

    /**
     * {@inheritDoc}
     */
    @Override
    public void deleteRow(EntityId entityId, long upToTimestamp) throws IOException {
        final State state = mState.get();
        Preconditions.checkState(state == State.OPEN,
            "Cannot delete row while TableWriter %s is in state %s.", this, state);

        final Delete delete = new Delete(entityId.getHBaseRowKey(), upToTimestamp);
        mHTable.delete(delete);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void deleteFamily(EntityId entityId, String family) throws IOException {
        deleteFamily(entityId, family, HConstants.LATEST_TIMESTAMP);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void deleteFamily(EntityId entityId, String family, long upToTimestamp)
        throws IOException {
        final State state = mState.get();
        Preconditions.checkState(state == State.OPEN,
            "Cannot delete family while TableWriter %s is in state %s.", this, state);

        // The only data in this ETS and DataCloud HBase family is the one cf family, so we can delete everything.
        final ColumnName columnName = ColumnName.create(family);
        final Delete delete = new Delete(entityId.getHBaseRowKey());
        delete.deleteFamily(columnName.getFamilyBytes(), upToTimestamp);

        // Send the delete to the HBase HTable.
        mHTable.delete(delete);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void deleteColumn(EntityId entityId, String family, String qualifier) throws IOException {
        deleteColumn(entityId, family, qualifier, HConstants.LATEST_TIMESTAMP);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void deleteColumn(EntityId entityId, String family, String qualifier, long upToTimestamp)
        throws IOException {
        final State state = mState.get();
        Preconditions.checkState(state == State.OPEN,
            "Cannot delete column while TableWriter %s is in state %s.", this, state);

        final ColumnName hbaseColumnName = ColumnName.create(family, qualifier);
        final Delete delete = new Delete(entityId.getHBaseRowKey())
            .deleteColumns(hbaseColumnName.getFamilyBytes(), hbaseColumnName.getQualifierBytes(), upToTimestamp);
        mHTable.delete(delete);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void deleteCell(EntityId entityId, String family, String qualifier) throws IOException {
        deleteCell(entityId, family, qualifier, HConstants.LATEST_TIMESTAMP);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void deleteCell(EntityId entityId, String family, String qualifier, long timestamp)
        throws IOException {
        final State state = mState.get();
        Preconditions.checkState(state == State.OPEN,
            "Cannot delete cell while TableWriter %s is in state %s.", this, state);

        final ColumnName hbaseColumnName = ColumnName.create(family, qualifier);
        final Delete delete = new Delete(entityId.getHBaseRowKey())
            .deleteColumn(hbaseColumnName.getFamilyBytes(), hbaseColumnName.getQualifierBytes(), timestamp);
        mHTable.delete(delete);
    }

    @Override
    public void flush() throws IOException {
        mHTable.flushCommits();
    }

    //endregion

    @Override
    public void close() throws IOException {
        final State oldState = mState.getAndSet(State.CLOSED);
        Preconditions.checkState(oldState == State.OPEN,
            "Cannot close TableWriter instance %s in state %s.", this, oldState);

        mHTable.close();
        mTable.release();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public String toString() {
        return Objects.toStringHelper(HBaseTableWriter.class)
            .add("id", System.identityHashCode(this))
            .add("table", mTable.getURI())
            .add("layout-name", mTableLayout.getDesc().getName())
            .add("state", mState)
            .toString();
    }

    /**
     * States of a writer instance.
     */
    private static enum State {
        UNINITIALIZED,
        OPEN,
        CLOSED
    }
    //endregion

    /**
     * Verifies that a column is a counter.
     *
     * @param family A column family.
     * @param qualifier A column qualifier.
     * @throws IOException If the column is not a counter, or it does not exist.
     */
    /*private void verifyIsCounter(String family, String qualifier) throws IOException {
        final ColumnName column = ColumnName.create(family, qualifier);
        if (mTableLayout.getCellSchema(column).getType() != SchemaType.COUNTER) {
            throw new IOException(String.format("Column '%s' is not a counter", column));
        }
    }*/

}
